
var pmx         = require('pmx');
var co          = require("co");
var chalk       = require("chalk");
var fs          = require("graceful-fs");
var axon        = require('axon');
var scheduler   = require('node-schedule');
var moment      = require("moment");
var format      = require("string-template")

var conf = pmx.initModule({
    // Options related to the display style on Keymetrics
    widget : {

        // Logo displayed
        logo : 'https://app.keymetrics.io/img/logo/keymetrics-300.png',

        // Module colors
        // 0 = main element
        // 1 = secondary
        // 2 = main border
        // 3 = secondary border
        theme            : ['#141A1F', '#222222', '#3ff', '#3ff'],

        // Section to show / hide
        el : {
            probes  : true,
            actions : true
        },

        // Main block to show / hide
        block : {
            actions : false,
            issues  : true,
            meta    : true,

            // Custom metrics to put in BIG
            main_probes : ['logs_lines', "logs_bytes"]
        }
    }
});

/**
 * Module specifics like connecting to a database and
 * displaying some metrics
 */

/**
 *                      Custom Metrics
 *
 * Let's expose some metrics that will be displayed into Keymetrics
 *   For more documentation about metrics: http://bit.ly/1PZrMFB
 */
var Probe = pmx.probe();

/**
 * .metric, .counter, .meter, .histogram are also available (cf doc)
 */
//收到的消息数
var messages = 0;
Probe.metric(
    {
        name : 'Messages',
        value : function() {
            return messages;
        }
    }
);

//收到的无效消息数
var messages_invalid = 0;
Probe.metric(
    {
        name : 'Messages Invalid',
        value : function() {
            return messages_invalid;
        }
    }
);

//日志总字节数
var logs_bytes = 0;
Probe.metric({
    name:   "Logs Bytes", 
    value:  function(){
        return logs_bytes;
    }
});

pmx.action('env', function(reply) {
    return reply(process.env);
});

function mkdir(path){
    if (fs.existsSync(path))
        return;

    let paths = path.split('/');
    for (let i = 1; i <= paths.length; ++i){
        let _path = paths.slice(0, i).join('/');
        if (!_path) continue;
        if (!fs.existsSync(_path))
            fs.mkdirSync(_path);
    }
}

var app_std_map = new Map();
co(function*(){
    const log_path = conf.log_path || "/runtimelog/app/{APPNAME}";

    var sock    = axon.socket("sub-emitter");
    var address = conf.address || "tcp://0.0.0.0:3000";
    sock.bind(address);
    sock.on('log:*', function(type, packet){
        ++messages;
        if (!(type == "out" || type == "err")){
            ++messages_invalid;
            return;
        }
        
        let data = packet.data;
        if (!data){
            ++messages_invalid;
            return;
        }

        let pm_proc = packet.process;
        let app_name= pm_proc.name;
        let app_id  = pm_proc.pm_id;
        let app     = app_std_map.get(app_name);
        if (!app){
            let appLogPath = format(log_path, {APPNAME: app_name});
            mkdir(appLogPath);

            let out = fs.createWriteStream(`${appLogPath}/${app_name}-out.log`, {flags: "a"});
            let err = fs.createWriteStream(`${appLogPath}/${app_name}-err.log`, {flags: "a"});
            app = {out, err};
            app_std_map.set(app_name, app);
        }

        let hostname= packet.host;
        app[type].write(chalk.bold.blue(`${hostname}(${app_id})`) + " | " + data);

        logs_bytes += data.length;
    });

    while(true) yield done => setTimeout(done, 50, null);
}).catch(e=>{
    console.error(e.stack || e);
}).then(()=>{
    process.exit(1);
});


/**
 * Apply the rotation process of the log file.
 *
 * @param {string} file 
 */
var DATE_FORMAT = 'YYYY-MM-DD_HH-mm-ss';
function proceed(file) {
    // set default final time
    var final_time = moment().format(DATE_FORMAT);
    var final_name = file.substr(0, file.length - 4) + '__' + final_time + '.log';

    // create our read/write streams
    var readStream  = fs.createReadStream(file);
    var writeStream = fs.createWriteStream(final_name, {'flags': 'w+'});

    // pipe all stream
    readStream.pipe(writeStream);
    
    // listen for error
    readStream.on('error', pmx.notify.bind(pmx));
    writeStream.on('error', pmx.notify.bind(pmx));

   // when the read is done, empty the file and check for retain option
    writeStream.on('finish', function() {
        readStream.close();
        writeStream.close();
        fs.truncate(file, function (err) {
            if (err)
                return pmx.notify(err);
            console.log('"' + final_name + '" has been created');
        });
    });
}


/**
 * Apply the rotation process if the `file` size exceeds the `SIZE_LIMIT`.
 * 
 * @param {string} file
 * @param {boolean} force - Do not check the SIZE_LIMIT and rotate everytime.
 */
function get_limit_size() {
    if (conf.max_size === '')
        return (1024 * 1024 * 10);
    if (typeof(conf.max_size) !== 'string')
        conf.max_size = conf.max_size + "";
    if (conf.max_size.slice(-1) === 'G')
        return (parseInt(conf.max_size) * 1024 * 1024 * 1024);
    if (conf.max_size.slice(-1) === 'M')
        return (parseInt(conf.max_size) * 1024 * 1024);
    if (conf.max_size.slice(-1) === 'K')
        return (parseInt(conf.max_size) * 1024);
    return parseInt(conf.max_size);
}
var SIZE_LIMIT = get_limit_size();
function proceed_file(file, force) {
    if (!fs.existsSync(file))
        return;

    fs.stat(file, function (err, data) {
        if (err) 
            return console.error(err);

        if (data.size > 0 && (data.size >= SIZE_LIMIT || force)) 
            proceed(file);
    });
}

function processAll(force = false){
    app_std_map.forEach((value, key)=>{
        for (let type in value){
            let std = value[type];
            let file_name = std.path;
            proceed_file(file_name, false);
        }
    });
}

// start background task
var WORKER_INTERVAL = 60 * 1000;
setInterval(processAll, WORKER_INTERVAL);

// register the cron to force rotate file
scheduler.scheduleJob("0 0 * * *", processAll.bind(null, true));